
#include <apr_strings.h>

#include "aprtools/APRAutoMemoryPool.h"
#include "aprtools/APRFile.h"
#include "iface/testerror.h"
#include "net/Broker5Client.h"
#include "net/BrokerMessage.h"
#include "utils/Hash.h"
#include "utils/StreamFromDirectory.h"
#include "utils/StreamToDirectory.h"

#include "acquire_common.h"
#include "acquire_common_private.h"
#include "acquire_client.h"
#include "acquire_compute.h"

char *globalplace_execname = (char *)"acquire";
char *globalplace_logname = (char *)"acquire.log";

acquireCommonGlobalState state;

static int isInitialised = 0;

/** Check to see if the system has been initialised correctly.
 * 
 * @return Non-zero on failure, with the code being checkable
 * here. Zero if all is fine. */
int acquireCommonStateIsIncomplete()
{
   int failed = 0x000;
   if (state.apr_iface == NULL)
   {
      errprint("the APR library has not been initialised");
      failed |= 0x0001;
   }
   if (state.ssl_iface == NULL)
   {
      errprint("openssl has not been initialised");
      failed |= 0x0002;
   }
   if (state.ssh2_iface == NULL)
   {
      errprint("libssh2 has not been initialised");
      failed |= 0x0004;
   }
   if (state.sqlite_iface == NULL)
   {
      errprint("the sqlite library has not been initialised");
      failed |= 0x0008;
   }
   if (state.keydb == NULL)
   {
      errprint("the key database has not been initialised");
      failed |= 0x0010;
   }
   if (state.operational_recipientkeyname == NULL)
   {
      errprint("have not found a working recipient key");
      failed |= 0x0020;
   }
   if (state.operational_senderkeyname == NULL)
   {
      errprint("have not found a working sender key");
      failed |= 0x0040;
   }
   if (state.my_name == NULL)
   {
      errprint("username is not set");
      failed |= 0x0080;
   }
   if (state.password == NULL)
   {
      errprint("password is not set");
      failed |= 0x0100;
   }
   if (state.hostname == NULL)
   {
      errprint("remote (broker) hostname is not set");
      failed |= 0x0200;
   }
   if (state.portno == 0)
   {
      errprint("remote (broker) port number is not set");
      failed |= 0x0400;
   }
   if (state.mp == NULL)
   {
      errprint("results memory pool has not been initialised");
      failed |= 0x0800;
   }
   return failed;
}

/** Initialising a common client state.
 *
 * @param hostname The host name of the broker server.
 *
 * @param portno The TCP port number of the broker server.
 *
 * @param senderkeyname The key (actor) to use for sending data.
 *
 * @param my_name The name of this actor.
 *
 * @param password The password required for opening the key chain.
 *
 * @param recipientkeyname The actor name of the recipient.
 * 
 * @param is_real_user Is the system a real 'user' client or a
 * 'compute' client?
 *
 * @return A registration state value: zero on error or failure,
 * non-zero denotes a code denoting a registration status. */
int acquireCommonInit(const char *hostname, int portno, const char *senderkeyname,
   const char *my_name, const char *password, const char *recipientkeyname, int is_real_user)
{
   acquireCommonInitLibraries();
   if (is_real_user)
   {
      return acquireClientInitState(hostname, portno, senderkeyname,
         my_name, password, recipientkeyname);
   } else
   {
      return acquireComputeInitState(hostname, portno, senderkeyname,
         my_name, password, recipientkeyname, NULL);
   }
}

/** Initialises the libraries required for the acquire client system
 * to operate. */
void acquireCommonInitLibraries()
{
   state.apr_iface = &(APRInterface::getInstance());
   state.ssl_iface = &(SSLInterface::getInstance());
   state.ssh2_iface = &(LibSSH2Interface::getInstance());
   state.sqlite_iface = &(SQLiteInterface::getInstance());
   state.mp = NULL;
   state.operational_recipientkeyname = NULL;
   state.operational_senderkeyname = NULL;
   state.my_name = NULL;
   state.hostname = NULL;
   state.portno = 0;
   state.keydb = NULL;
}

/** Destroy the held state that encodes a client context. */
void acquireCommonDestroyState()
{
   if (acquireCommonTestInitialised() == 0) return;
   isInitialised = 0;
   if (state.my_name) free((void *)state.my_name);
   state.my_name = NULL;
   if (state.password)
   {
      memset((void *)state.password, 0, strlen(state.password));
      free((void *)state.password);
   }
   state.password = NULL;
   if (state.hostname) free((void *)state.hostname);
   state.hostname = NULL;
   if (state.operational_recipientkeyname) free((void *)state.operational_recipientkeyname);
   state.operational_recipientkeyname = NULL;
   if (state.operational_senderkeyname) free((void *)state.operational_senderkeyname);
   state.operational_senderkeyname = NULL;
   if (state.keydb) delete state.keydb;
   state.keydb = NULL;
   if (state.mp) apr_pool_destroy(state.mp);
   state.mp = NULL;
}

/** Destroy the contexts in the required libraries. */
void acquireCommonDestroy()
{
   acquireCommonDestroyState();
   if (state.sqlite_iface) delete state.sqlite_iface;
   state.sqlite_iface = NULL;
   if (state.ssh2_iface) delete state.ssh2_iface;
   state.ssh2_iface = NULL;
   if (state.ssl_iface) delete state.ssl_iface;
   state.ssl_iface = NULL;
   if (state.apr_iface) delete state.apr_iface;
   state.apr_iface = NULL;
}

/** Clear the memory pool that holds any permanent results. */
void acquireCommonClearResults()
{
   if (acquireCommonStateIsIncomplete()) return;
   apr_pool_t *oldmp = state.mp;
   state.mp = NULL;
   apr_status_t status;
   status = apr_pool_create(&(state.mp), (state.apr_iface)->getGlobalPool());
   CheckAPRError(status);
   if (state.mp == NULL)
   {
      errprint("we do not have a memory pool for results");
      state.mp = oldmp;
   } else
   {
      apr_pool_destroy(oldmp);
   }
}

/** Test if the system is initialised.
 * 
 * @return One if initialised, zero if not initialised. */
int acquireCommonTestInitialised()
{
   return isInitialised;
}

/** Set the system to be in the initialised state. */
void acquireCommonSetInitialised()
{
   isInitialised = 1;
}

/** Client allocation of data store. This uses compression and
 * encryption to determine the size of the eventual data and sends the
 * hashes of the unencrypted blocks and sizes of the encrypted blocks
 * to the server. This is so that the server can allocate space, and
 * later find out if the client copy desynchronises through a
 * comparison of the hash (it should not rehash as this would require
 * data decryption).
 * 
 * @param inputdirectory Input directory for the data store. Data
 * streams from here.
 * 
 * @param enckeyname Allows the caller to override the default key used
 * to encrypt/decrypt the data store with their own specfied key without
 * creating and distributing a new key.
 * 
 * @param deckeyname Allows the caller to override the default key used
 * to encrypt/decrypt any child data stores with their own specfied key
 * without creating and distributing a new key.
 * 
 * @param addspecifickeys Switch to create and add data store specific
 * keys to this datastore.
 * 
 * @param tblocksize Unencrypted block size.
 * 
 * @param remotespace Allows the updated quota amount, after the
 * preallocation of the directory data remotely to be reflected in the
 * return value to the client.
 *
 * @param blocks The number of blocks that the data in the directory
 * was split into.
 * 
 * @param cancel Cancellation from another thread changes the integer
 * that this points to, enabling an asynchronous cancellation.
 * 
 * @return The hexadecimal ID string of the data store if
 * successful. On failure, this is simply NULL. The string returned is
 * constant because it is held in the results memory pool, which can
 * be cleared using acquireCommonClearResults(). */
const char *acquireCommonCreateDataStore(const char *inputdirectory,
   const char *enckeyname, const char *deckeyname,
   int addspecifickeys, int tblocksize, int64_t *remotespace, int *blocks, int *cancel)
{
   int cancelled = 0;
   if (acquireCommonStateIsIncomplete()) return NULL;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   bmsg.setRecipient(state.operational_recipientkeyname);
   bmsg.setSender(state.operational_senderkeyname);
   bmsg.addTypeHeaderNOCRLF("START_DATA_STORE", state.my_name);
   bmsg.addHeaderLineNOCRLF("Unencrypted-Block-Size: %d", tblocksize);
   char *hexid = StringUtils::replaceWithPoolString(Hash::makeHexadecimalID(32), pool.mp());
   char *retval = NULL;
   char *input_key_name = StringUtils::mpprintf(pool.mp(), "%s_inp", hexid);
   if (enckeyname) input_key_name = apr_pstrdup(pool.mp(), enckeyname);
   char *output_key_name = StringUtils::mpprintf(pool.mp(), "%s_out", hexid);
   if (deckeyname) output_key_name = apr_pstrdup(pool.mp(), deckeyname);
   bmsg.addHeaderLineNOCRLF("Data-Store-ID: %s", hexid);
   bmsg.addHeaderLineNOCRLF("Data-Store-Decryption-Key: %s", input_key_name);
   // Key pairs to create the temporary job actor
   (state.keydb)->generateEncDecPairIntoDB(false, false, hexid, state.my_name, AbstractKey::KEY_SYMMETRIC_AES);
   (state.keydb)->generateSgnChkPairIntoDB(false, false, false, hexid, state.my_name, AbstractKey::KEY_ASYMMETRIC_PAIR_ECDSA);
   // Attach all the keys for the job, as we are creating a remote
   // actor for this with keys
   bmsg.insertAllKeysToSend(0, hexid, NULL, state.keydb);
   if (addspecifickeys)
   {
      // This is local, so we don't have to set key dependencies -
      // this is for the input to the cluster (workpacket)
      if (enckeyname == NULL)
         (state.keydb)->generateEncDecPairIntoDB(false, false, input_key_name, state.my_name, AbstractKey::KEY_ASYMMETRIC_PAIR_RSA);
      // This is remote, so we don't have to set key dependencies -
      // this is for the output from the cluster (results)
      if (deckeyname == NULL)
         (state.keydb)->generateEncDecPairIntoDB(false, false, output_key_name, state.my_name, AbstractKey::KEY_ASYMMETRIC_PAIR_RSA);
      // Attach secret keys for the decryption of the workpacket
      bmsg.insertSecretKeysToSend(1, input_key_name, NULL, state.keydb);
      // Attach available keys for the encryption of the results in
      // the other direction
      bmsg.insertAvailableKeysToSend(2, output_key_name, NULL, state.keydb);
      // Attach available keys for the encryption of further
      // workpackets
      bmsg.insertAvailableKeysToSend(3, input_key_name, NULL, state.keydb);
      bmsg.addHeaderLineNOCRLF("WorkPacket-Decryption-Key-Slot: 1");
      bmsg.addHeaderLineNOCRLF("Results-Encryption-Key-Slot: 2");
      bmsg.addHeaderLineNOCRLF("WorkPacket-Encryption-Key-Slot: 3");
   }
   EncryptionKey *enc_wrap_input_key = Key::getEncryptionKey(state.keydb, NULL, input_key_name);
   if (enc_wrap_input_key)
   {
      StreamFromDirectory s_from_dir(inputdirectory, pool.mp());
      EncryptWrapper enc_wrap_input(enc_wrap_input_key, NULL, true);
      int plain_size = tblocksize;
      std::vector<char *> hashlist;
      std::vector<char *> cesizes;
      *blocks = 0;
      char *block_data_buffer = (char *)malloc(sizeof(char)*tblocksize);
      while (plain_size == tblocksize)
      {
         char *thash = NULL;
         int compressed_encrypted_size = 0;
         if (cancel)
            if (*cancel)
            {
               cancelled = 1;
               break;
            }
         char *this_block = s_from_dir.getNextBlockEncrypted(&enc_wrap_input, &plain_size, &compressed_encrypted_size, &thash);
         hashlist.push_back(thash);
         thash = NULL;
         cesizes.push_back(StringUtils::varprintf("%d", compressed_encrypted_size));
         free(this_block);
         (*blocks)++;
      }
      if (cancel)
         if (*cancel)
            cancelled = 1;
      if (!cancelled)
      {
         char *all_hashes = StringUtils::flattenStringList(&hashlist, ",");
         char *all_sizes = StringUtils::flattenStringList(&cesizes, ",");
         // uses special destroyStringList to destroy the string entries.
         StringUtils::destroyStringList(hashlist);
         StringUtils::destroyStringList(cesizes);
         int hashes_payload_number = bmsg.addVerbatimPayloadString(all_hashes);
         int sizes_payload_number = bmsg.addVerbatimPayloadString(all_sizes);
         bmsg.addHeaderLineNOCRLF("Payload-Of-Hashes: %d", hashes_payload_number);
         bmsg.addHeaderLineNOCRLF("Payload-Of-Sizes: %d", sizes_payload_number);
         free(all_hashes);
         free(all_sizes);
         if (exchangeMessage(bmsg, pool.mp(), NULL) == 1)
         {
            errprint("message exchange failed");
            if (enc_wrap_input_key) delete enc_wrap_input_key;
            return NULL;
         }
         if (bmsg.getProcessee())
         {
            if (strcmp(bmsg.getType(), "DATA_STORE_DENIED") == 0)
            {
               if (strcmp(bmsg.getProcessee(), "REQUEST_INVALID") == 0)
                  errprint("server reports invalid request");
               if (strcmp(bmsg.getProcessee(), "USER_HAS_NO_RESOURCES") == 0)
                  errprint("user has no resources - must register an SSH login");
               if (strcmp(bmsg.getProcessee(), "ALREADY_EXISTS") == 0)
                  errprint("data store already exists");
               if (strcmp(bmsg.getProcessee(), "OUT_OF_SPACE") == 0)
                  errprint("user quota has run out - cannot allocate this new datastore");
            }
            if (strcmp(bmsg.getType(), "DATA_STORE_CREATED") == 0)
            {
               // need to get new quota size to update it locally
               retval = apr_pstrdup(state.mp, hexid);
            }
         }
      }
      delete enc_wrap_input_key;
   }
   // if the return value isn't the data store ID then it is NULL and
   // if it is NULL, then we don't need the data store ID anymore
   // (doesn't exist!).
   return (const char *)retval;
}

/** Download and reconstruct data from the data store on the broker.
 * 
 * @param directory The directory into which the files should be
 * reconstructed.
 * 
 * @param blockcount The number of blocks in the data store which must
 * be downloaded and reconstructed.
 * 
 * @param datastoreid The data store ID or name that must be
 * downloaded to the local disk.
 * 
 * @param cancelflag A pointer to an area of memory which is
 * dirty-read to provide a instant cancel to an external thread.
 * 
 * @param decrypterkey The name of the key to be used to decrypt the
 * downloaded datastore.
 * 
 * @return Zero on success, non-zero on failure. */
int acquireCommonDownloadFromDataStore(const char *directory, int blockcount,
   const char *datastoreid, int *cancelflag, const char *decrypterkey)
{
   if (acquireCommonStateIsIncomplete()) return 1;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   StreamToDirectory std(directory, pool.mp());
   for (int i = 0; i < blockcount; i++)
   {
      int retries = 5;
      int block_unfinished = 1;
      while ((retries) && (block_unfinished))
      {
         bmsg.reset();
         bmsg.setRecipient(state.operational_recipientkeyname);
         bmsg.setSender(state.operational_senderkeyname);
         bmsg.addTypeHeaderNOCRLF("GET_DATA_BLOCK", datastoreid);
         bmsg.addHeaderLineNOCRLF("Block-ID: %d", i);
         if (exchangeMessage(bmsg, pool.mp(), cancelflag) == 1)
         {
            errprint("message exchange failed");
            return 1;
         }
         if ((bmsg.getType()) && (strcmp(bmsg.getType(), "DOWNLOAD_DATA_BLOCK") == 0))
         {
            const char *payload_number_string = (bmsg.getHeaderTable())->get("Block-At-Payload");
            const char *payload_real_hash = (bmsg.getHeaderTable())->get("Block-Real-Hash");
            if ((payload_number_string) && (strlen(payload_number_string) > 0) &&
                (payload_real_hash) && (strlen(payload_real_hash) > 0))
            {
               int payload_no = atoi(payload_number_string);
               char *hex_hash = Hash::getHexSHA1Hash(bmsg.getPayload(payload_no), bmsg.getPayloadLength(payload_no), NULL);
               if (strcmp(payload_real_hash, hex_hash) != 0)
               {
                  errprint("hash of downloaded block %d failed, retrying", i);
               } else
               {
                  apr_off_t realsize = -1;
                  char *hash_data = NULL;
                  DecryptWrapper *decwrap = (state.keydb)->getDecrypter(decrypterkey, NULL, true);
                  int retval = std.putNextBlockDecrypted(decwrap, bmsg.getPayload(payload_no),
                     &realsize, bmsg.getPayloadLength(payload_no), &hash_data);
                  delete decwrap;
                  const char *payload_unenc_hash = (bmsg.getHeaderTable())->get("Block-Plain-Hash");
                  if (retval == 1)
                  {
                     if (payload_unenc_hash)
                     {
                        if (strcmp(payload_unenc_hash, hash_data) != 0)
                        {
                           errprint("hash of plain decrypted data from block %d failed", i);
                        } else
                        {
                           errprint("successfully downloaded block %d", i);
                           block_unfinished = 0;
                        }
                     } else
                     {
                        errprint("could not find plain text hash for block %d", i);
                     }
                  } else if (retval == 0)
                  {
                     errprint("size mismatch when streaming in block %d", i);
                  } else
                  {
                     errprint("error occurred putting block %d into directory", i);
                  }
               }
            } else
            {
               errprint("corruption or error in message format");
            }
         } else
         {
            errprint("data block download failed");
         }
         if (block_unfinished == 1)
            retries--;
      }
      if (retries == 0)
      {
         errprint("ran out of retries on block %d of work instance %s on data store %s", i, state.my_name, datastoreid); 
         return 1;
      }
   }
   return 0;
}

/** Upload data to the data store from the given directory. The
 * contents of the directory must be the same as what was promised to
 * the broker earlier. No state is used for this.
 * 
 * @param store_id The ID of the work store to upload to.
 * 
 * @param inputdirectory The directory from which the data will come.
 * 
 * @param tblocksize The block size of the blocks to upload.
 * 
 * @param blocksdone The number of blocks that have so far completed
 * uploading.
 * 
 * @param cancelflag A cancel flag to enable asynchronous cancellation
 * of the uploading operation.
 * 
 * @param realkeyname The name ofthe key to use to encrypt the data
 * being uploaded.
 * 
 * @return A value denoting whether the upload succeeded. If it is
 * zero, then no error occurred, otherwise an error flag will be
 * returned here. */
int acquireCommonUploadToDataStore(const char *store_id, const char *inputdirectory, int tblocksize,
   int *blocksdone, int *cancelflag, const char *realkeyname)
{
   int timeout = 1;
   int cancelled = 0;
   if (acquireCommonStateIsIncomplete()) return 1;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   EncryptionKey *enc_wrap_input_key = Key::getEncryptionKey(state.keydb, NULL, realkeyname);
   if (enc_wrap_input_key)
   {
      int blocks_curr = 0;
      int blocks_remaining = 1;
      {
         EncryptWrapper enc_wrap_input(enc_wrap_input_key, NULL, true);
         while (blocks_remaining > 0)
         {
            StreamFromDirectory s_from_dir(inputdirectory, pool.mp());
            if (cancelflag)
               if (*cancelflag)
               {
                  cancelled = 1;
                  break;
               }
            int plain_size = tblocksize;
            int blockno = 0;
            while (plain_size == tblocksize)
            {
               bmsg.reset();
               int compressed_encrypted_size = 0;
               char *thash = NULL;
               char *this_block = s_from_dir.getNextBlockEncrypted(&enc_wrap_input, &plain_size, &compressed_encrypted_size, &thash);
               bmsg.setRecipient(store_id);
               bmsg.setSender(store_id);
               bmsg.addTypeHeaderNOCRLF("CHECK_DATA_BLOCK", store_id);
               bmsg.addHeaderLineNOCRLF("Block-ID: %d", blockno);
               bmsg.addHeaderLineNOCRLF("Block-Plain-Hash: %s", thash);
               bmsg.addHeaderLineNOCRLF("Block-Size: %d", compressed_encrypted_size);
               if (exchangeMessage(bmsg, pool.mp(), cancelflag) == 1)
               {
                  errprint("message exchange failed");
                  if (enc_wrap_input_key) delete enc_wrap_input_key;
                  return 1;
               }
               int canskip = 1;
               //bmsg.print();
               if (bmsg.isType("DATA_BLOCK_STATUS"))
               {
                  canskip = -1;
                  if (bmsg.isProcessee("DATA_BLOCK_INCOMPLETE")) canskip = 0;
                  if (bmsg.isProcessee("DATA_BLOCK_COMPLETE")) { canskip = 1; blocks_curr++; }
                  if (canskip == -1)
                  {
                     bmsg.print();
                     errprint("work store id %s, block %d, server tells us this is an invalid block (%s)!\n", store_id, blockno, bmsg.getProcessee());
                  }
               }
               if (canskip == 0)
               {
                  bmsg.reset();
                  bmsg.setRecipient(store_id); // should probably be store_id!
                  bmsg.setSender(store_id);
                  bmsg.addTypeHeaderNOCRLF("UPLOAD_DATA_BLOCK", store_id);
                  //bmsg->addHeaderLineNOCRLF("Data-Store-ID: %s", store_id);
                  bmsg.addHeaderLineNOCRLF("Block-ID: %d", blockno);
                  bmsg.addHeaderLineNOCRLF("Block-Size: %d", compressed_encrypted_size);
                  if (thash)
                     bmsg.addHeaderLineNOCRLF("Block-Plain-Hash: %s", thash);
                  char *real_hash = Hash::getHexSHA1Hash(this_block, compressed_encrypted_size, NULL);
                  bmsg.addHeaderLineNOCRLF("Block-Real-Hash: %s", real_hash);
                  free(real_hash);
                  int payload_no = bmsg.addPayloadBuffer(this_block, compressed_encrypted_size);
                  bmsg.addHeaderLineNOCRLF("Block-Data-Payload: %d", payload_no);
                  exchangeMessage(bmsg, pool.mp(), cancelflag);
                  if ((bmsg.isType("DATA_BLOCK_STATUS")) && (bmsg.isProcessee("DATA_BLOCK_COMPLETE")))
                  {
                     blocks_curr++;
                  }
               }
               free(thash);
               free(this_block);
               blockno++;
            }
            blocks_remaining = blockno - blocks_curr;
            blocks_curr = 0;
         }
      }
      int retval = 1;
      if (blocks_remaining == 0) retval = 0;
      delete enc_wrap_input_key;
      return retval;
   } else
   {
      errprint("could not get encryption key '%s'\n", realkeyname);
   }
   return 1;
}

/** Download data from the data store on the broker.
 * 
 * @param datastoreid The data store ID or name that must be
 * downloaded to the local disk.
 * 
 * @param directory The directory into which the temporary encrypted
 * block files should be placed.
 * 
 * @param blockcount The number of blocks in the data store which must
 * be downloaded.
 * 
 * @param cancelflag A pointer to an area of memory which is
 * dirty-read to provide a instant cancel to an external thread.
 * 
 * @return Zero on success, non-zero on failure. */
int acquireCommonDeferredDownloadFromDataStore(const char *datastoreid,
   const char *directory, int blockcount, int *cancelflag)
{
   if (acquireCommonStateIsIncomplete()) return 1;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   for (int i = 0; i < blockcount; i++)
   {
      int retries = 5;
      int block_unfinished = 1;
      while ((retries) && (block_unfinished))
      {
         bmsg.reset();
         bmsg.setRecipient(state.operational_recipientkeyname);
         bmsg.setSender(state.operational_senderkeyname);
         bmsg.addTypeHeaderNOCRLF("GET_DATA_BLOCK", datastoreid);
         bmsg.addHeaderLineNOCRLF("Block-ID: %d", i);
         if (exchangeMessage(bmsg, pool.mp(), cancelflag) == 1)
         {
            errprint("message exchange failed");
            return 1;
         }
         if ((bmsg.getType()) && (strcmp(bmsg.getType(), "DOWNLOAD_DATA_BLOCK") == 0))
         {
            const char *payload_number_string = (bmsg.getHeaderTable())->get("Block-At-Payload");
            const char *payload_real_hash = (bmsg.getHeaderTable())->get("Block-Real-Hash");
            if ((payload_number_string) && (strlen(payload_number_string) > 0) &&
                (payload_real_hash) && (strlen(payload_real_hash) > 0))
            {
               int payload_no = atoi(payload_number_string);
               char *hex_hash = Hash::getHexSHA1Hash(bmsg.getPayload(payload_no), bmsg.getPayloadLength(payload_no), NULL);
               if (strcmp(payload_real_hash, hex_hash) != 0)
               {
                  errprint("hash of downloaded block %d failed, retrying", i);
               } else
               {
                  apr_off_t realsize = -1;
                  char *hash_data = NULL;
                  char *tfilename = StringUtils::varprintf("%s/datastore_%s.block%d", directory, datastoreid, i);
                  int retval = StringUtils::writeBufferToFile(bmsg.getPayload(payload_no),
                     bmsg.getPayloadLength(payload_no), tfilename, pool.mp());
                  free(tfilename);
                  const char *payload_unenc_hash = (bmsg.getHeaderTable())->get("Block-Plain-Hash");
                  if (retval != 0)
                  {
                     errprint("error occurred putting block %d into file '%s'", i, tfilename);
                  } else block_unfinished = 0;
                  // unfortunately, we can't check the plain data
                  // integrity now (unless we do the decryption now?)
                  // - the encrypted data integrity should be enough
               }
            } else
            {
               errprint("corruption or error in message format");
            }
         } else
         {
            errprint("data block download failed");
         }
         if (block_unfinished == 1)
            retries--;
      }
      if (retries == 0)
      {
         errprint("ran out of retries on block %d of work instance %s on data store %s", i, state.my_name, datastoreid); 
         return 1;
      }
   }
   return 0;
}

/** Given that the deferred download has saved the data store into
 * blocks in temporary disk storage, this function can now reconstruct
 * the real directory of result files from the block files using a
 * decryption key.
 * 
 * @param store_id The data store name of the store that files were
 * downloaded from.
 * 
 * @param tempdirectory The temporary directory that the files were
 * saved into.
 * 
 * @param outputdirectory The output directory that the reconstructed
 * files should be output into.
 * 
 * @param blockcount The number of the blocks that have been
 * downloaded from the data store and need to be reconstructed into
 * the directory.
 * 
 * @param realkeyname The real key name of the decrypting key. If the
 * key does not exist locally, the broker will be queried to see if
 * the name of the key has been changed during the chain of
 * computations.
 * 
 * @return Returns zero on success, non-zero on failure. */
int acquireCommonReconstructDeferredDownload(const char *store_id,
   const char *tempdirectory, const char *outputdirectory, int blockcount, const char *realkeyname)
{
   if (acquireCommonStateIsIncomplete()) return 1;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   StreamToDirectory std(outputdirectory, pool.mp());
   const char *newkeyname = realkeyname;
   // if the key given by the result block is not found, contact the
   // broker to find alternate keys
   while (!((state.keydb)->decryptionTable()->testKeyExists(newkeyname)))
   {
      BrokerMessage bmsg(pool.mp());
      bmsg.setRecipient(state.operational_recipientkeyname);
      bmsg.setSender(state.operational_senderkeyname);
      bmsg.addTypeHeaderNOCRLF("RESOLVE_KEY_EQUIVALENCE", newkeyname);
      bmsg.print();
      if (exchangeMessage(bmsg, pool.mp(), NULL) == 1)
      {
         errprint("message exchange failed");
         return 1;
      }
      bmsg.print();
      if ((bmsg.isType("ALTERNATIVE_KEY")) && (bmsg.getProcessee()))
      {
         newkeyname = apr_pstrdup(pool.mp(), bmsg.getProcessee());
      } else break;
   }
   errprint("newkeyname: %s", newkeyname);
   for (int i = 0; i < blockcount; i++)
   {
      char *tfilename = StringUtils::varprintf("%s/datastore_%s.block%d", tempdirectory, store_id, i);
      char *databuffer = NULL;
      int databufferlen = 0;
      int retcode = StringUtils::readFileToBuffer(&databuffer, &databufferlen, tfilename, pool.mp());
      apr_off_t realsize = -1;
      char *hash_data = NULL;
      DecryptWrapper *decwrap = (state.keydb)->getDecrypter(newkeyname, NULL, true);
      int retval = std.putNextBlockDecrypted(decwrap, databuffer,
         &realsize, databufferlen, &hash_data);
      delete decwrap;
      free(databuffer);
      apr_status_t status = APRFile::removeWrapper(tfilename, pool.mp());
      CheckAPRError(status);
   }
   return 0;
}

/** Set purpose flags on a completed data store. This is done by the
 * uploading process so that no incomplete data stores are accidently
 * marked as complete. If the data is to be marked as a work packet,
 * then other fields are required to define the work instances that
 * would result.
 *
 * @param datastoreid The data store name to purpose (or repurpose).
 * 
 * @param isworkpacket Is the data store to be marked as a workpacket?
 * (1 - yes, 0 - no)
 * 
 * @param isresults Is the data store to be marked as a set of
 * results? (1 - yes, 0 - no)
 * 
 * @param clusterid The cluster ID or cluster list to be used for this
 * workpacket. NULL can be used if the data store is not to be a work
 * packet.
 * 
 * @param maxtimesecs If the data store is to be a work packet, what
 * is the maximum time to be allowed for one of its instances to run?
 * 
 * @param instances If the data store is a work packet, how many
 * instances of work should it spawn?
 * 
 * @param ipassword If the data store is a work packet, it might
 * require an SSH deferred keychain password to resubmit further work
 * for extended processing. The required password should be specified
 * here.
 * 
 * @return One on success, zero on failure, minus one on failed to start work,
 * minus two on unknown error. */
int acquireCommonSetPurposeFlagsOnDataStore(const char *datastoreid, int isworkpacket, int isresults,
   const char *clusterid, int maxtimesecs, int instances, const char *ipassword)
{
   if (acquireCommonStateIsIncomplete()) return ACQUIRE_PURPOSE_RESULT__ERROR;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   bmsg.setSender(datastoreid);
   bmsg.setRecipient(datastoreid);
   bmsg.addTypeHeaderNOCRLF("PURPOSE_DATASTORE", datastoreid);
   bmsg.addHeaderLineNOCRLF("Complete: true");
   if (isresults)
      bmsg.addHeaderLineNOCRLF("Results: true");
   if (isworkpacket)
   {
      bmsg.addHeaderLineNOCRLF("Work: true");
      if ((clusterid != NULL) && (strlen(clusterid) > 0))
         bmsg.addHeaderLineNOCRLF("Cluster-IDs: %s", clusterid);
      bmsg.addHeaderLineNOCRLF("Max-Time: %d", maxtimesecs);
      bmsg.addHeaderLineNOCRLF("Instances: %d", instances);
      if (ipassword)
      {
         bmsg.addHeaderLineNOCRLF("Secret-Password: %s", ipassword);
      } else if (state.password)
      {
         bmsg.addHeaderLineNOCRLF("Secret-Password: %s", state.password);
      }
   }
   if (exchangeMessage(bmsg, pool.mp(), NULL) == 1)
   {
      errprint("message exchange failed");
      return ACQUIRE_PURPOSE_RESULT__ERROR;
   }
   if (bmsg.isProcessee("DATA_STORE_INCOMPLETE"))
   {
      errprint("datastore not complete (maybe execution failed/no clusters?)");
      return ACQUIRE_PURPOSE_RESULT__FAILED;
   }
   if (bmsg.isProcessee("DATA_STORE_COMPLETE"))
   {
      return ACQUIRE_PURPOSE_RESULT__SUCCESS;
   }
   errprint("could not interpret the result from purposing the datastore");
   return ACQUIRE_PURPOSE_RESULT__UNKNOWN;
}

/** Delete a remote data store.
 * 
 * @param hex_store_id The hexadecimal ID string of the data store.
 * 
 * @return An error code, zero if successful. Failure probably
 * ignorable, given that it's a delete function. */
int acquireCommonDeleteDataStore(const char *hex_store_id)
{
   if (acquireCommonStateIsIncomplete()) return -1;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   bmsg.setRecipient(state.operational_recipientkeyname);
   bmsg.setSender(state.operational_senderkeyname);
   bmsg.addTypeHeaderNOCRLF("DELETE_DATA_STORE", state.my_name);
   bmsg.addHeaderLineNOCRLF("Data-Store-ID: %s", hex_store_id);
   if (exchangeMessage(bmsg, pool.mp(), NULL) == 1)
   {
      errprint("message exchange failed");
      return -1;
   }
   int retval = -1;
   if (bmsg.isProcessee("DATA_STORE_DELETE_SUCCESS")) retval = ACQUIRE_DELETE_DATA_STORE__SUCCESS;
   if (bmsg.isProcessee("DATA_STORE_DELETE_DENIED")) retval = ACQUIRE_DELETE_DATA_STORE__DENIED;
   return retval;
}

/** Get the parent of the given actor.
 * 
 * @param actorname The name of the actor to get the parent of.
 * 
 * @return The parent of the given actor. The string returned is
 * constant because it is held in the results memory pool, which can
 * be cleared using acquireCommonClearResults(). */
const char *acquireCommonGetParent(const char *actorname)
{
   if ((actorname == NULL) || (strlen(actorname) > 0)) return NULL;
   if (acquireCommonStateIsIncomplete()) return NULL;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   bmsg.setRecipient(state.operational_recipientkeyname);
   bmsg.setSender(state.operational_senderkeyname);
   bmsg.addTypeHeaderNOCRLF("GET_PARENT", actorname);
   if (exchangeMessage(bmsg, pool.mp(), NULL) == 1)
   {
      errprint("message exchange failed");
      return NULL;
   }
   if (bmsg.getType())
   {
      if ((strcmp(bmsg.getType(), "PARENT_DATA") == 0) &&
          (bmsg.getPayload(0)) && (bmsg.getPayloadLength(0) > 0))
         return apr_pstrdup(state.mp, bmsg.getPayload(0));
      if (strcmp(bmsg.getType(), "NO_PARENT") == 0)
         return apr_pstrdup(state.mp, "");
   }
   return NULL;
}

/** Get the children of the given actor.
 * 
 * @param actorname The name of the actor to get the children of.
 * 
 * @return The children of the given actor. The string list returned
 * is constant because it is held in the results memory pool, which
 * can be cleared using acquireCommonClearResults(). */
const char **acquireCommonGetChildren(const char *actorname)
{
   if ((actorname == NULL) || (strlen(actorname) == 0)) return NULL;
   if (acquireCommonStateIsIncomplete()) return NULL;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   bmsg.setRecipient(state.operational_recipientkeyname);
   bmsg.setSender(state.operational_senderkeyname);
   bmsg.addTypeHeaderNOCRLF("GET_CHILDREN", actorname);
   if (exchangeMessage(bmsg, pool.mp(), NULL) == 1)
   {
      errprint("message exchange failed");
      return NULL;
   }
   if (bmsg.getType())
   {
      if ((strcmp(bmsg.getType(), "CHILDREN_DATA") == 0) &&
          (bmsg.getPayload(0)) && (bmsg.getPayloadLength(0) > 0))
      {
         std::vector<char *> *childlist = StringUtils::unflattenStringList(bmsg.getPayload(0), ",");
         char **childlist_mp = (char **)apr_palloc(state.mp, sizeof(char *)*(childlist->size() + 1));
         for (int i = 0; i < childlist->size(); i++)
            childlist_mp[i] = apr_pstrdup(state.mp, childlist->at(i));
         childlist_mp[childlist->size()] = NULL;
         StringUtils::destroyStringList(childlist);
         return (const char **)childlist_mp;
      }
      if (strcmp(bmsg.getType(), "NO_CHILDREN") == 0)
      {
         char **childlist_mp = (char **)apr_palloc(state.mp, 1);
         childlist_mp[0] = NULL;
         return (const char **)childlist_mp;
      }
   }
   return NULL;
}

/** Returns a structure containing all of the information returned
 * about the data store.
 * 
 * @param datastoreid The data store name and ID.
 * 
 * @return The structure containing all of the data store
 * information. */
acquireCommonDataStoreInfo acquireCommonGetDataStoreInformation(const char *datastoreid)
{
   acquireCommonDataStoreInfo dsinfo;
   dsinfo.name = datastoreid;
   dsinfo.isvalid = 0;
   if (acquireCommonStateIsIncomplete()) return dsinfo;
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   bmsg.setRecipient(state.operational_recipientkeyname);
   bmsg.setSender(state.operational_senderkeyname);
   bmsg.addTypeHeaderNOCRLF("CHECK_DATA_STORE", datastoreid);
   if (exchangeMessage(bmsg, pool.mp(), NULL) == 1)
   {
      errprint("message exchange failed");
      return dsinfo;
   }
   const APRTable *headertable = bmsg.getHeaderTable();
   if ((headertable->get("DS-Decryption-Key")) == NULL)
   {
      errprint("could not find name of the decrypting key in the DATA_STORE_STATUS (type=%s, processee=%s)", bmsg.getType(), bmsg.getProcessee());
      return dsinfo;
   }
   dsinfo.decryptionkey = (const char *)apr_pstrdup(state.mp, headertable->get("DS-Decryption-Key"));
   if ((headertable->get("DS-Blocks")) == NULL)
   {
      errprint("could not find number of blocks in DATA_STORE_STATUS (type=%s, processee=%s)", bmsg.getType(), bmsg.getProcessee());
      return dsinfo;
   }
   dsinfo.blockcount = atoi(headertable->get("DS-Blocks"));
   if ((headertable->get("DS-BlockSize")) == NULL)
   {
      errprint("could not find unencrypted block size in DATA_STORE_STATUS (type=%s, processee=%s)", bmsg.getType(), bmsg.getProcessee());
      return dsinfo;
   }
   dsinfo.blocksize = atoi(headertable->get("DS-BlockSize"));
   if ((headertable->get("DS-Instances")) == NULL)
   {
      errprint("could not find number of instances in DATA_STORE_STATUS (type=%s, processee=%s)", bmsg.getType(), bmsg.getProcessee());
      return dsinfo;
   }
   dsinfo.instancecount = atoi(headertable->get("DS-Instances"));
   if ((headertable->get("DS-WorkPacket")) == NULL)
   {
      errprint("could not find WorkPacket switch (tried to run an unrunnable data store?) DATA_STORE_STATUS (type=%s, processee=%s)", bmsg.getType(), bmsg.getProcessee());
      return dsinfo;
   }
   dsinfo.isworkpacket = (((headertable->get("DS-WorkPacket")) != NULL) &&
                          (strcmp(headertable->get("DS-WorkPacket"), "true") == 0));
   dsinfo.isresults    = (((headertable->get("DS-Results")) != NULL) &&
                          (strcmp(headertable->get("DS-Results"), "true") == 0));
   dsinfo.iscomplete   = ((bmsg.getProcessee() != NULL) &&
                          (strcmp(bmsg.getProcessee(), "DATA_STORE_COMPLETE") == 0));
   dsinfo.isvalid = 1;
   return dsinfo;
}

/** Resolves a human-readable name using the remote name resolution service.
 * 
 * @param hierarchical
 * The human-readable hierarchical namespacing name to resolve.
 * 
 * @return
 * The string denoting the UID or internal name of the referenced object. */
const char *acquireCommonResolve(const char *hierarchical)
{
   return apr_pstrdup(state.mp, (state.resolver)->get(hierarchical));
}

/** Renames an ID, which is not human-readable, to something human readable
 * in the hierarchy. Does this by changing the external name of the actor.
 * 
 * @param actorid
 * The actor ID name of the actor whose human-readable name is to be changed.
 * 
 * @param actoralias
 * The new alias of the actor.
 * 
 * @return
 * Zero if successful, one if the name is invalid, less than zero if
 * unsuccessful. */
int acquireCommonRenameActor(const char *actorid, const char *actoralias)
{
   FATAL_IF_BLANK(-1, actorid, "actor id name");
   FATAL_IF_BLANK(-1, actoralias, "actor alias name");
   APRAutoMemoryPool pool((state.apr_iface)->getGlobalPool());
   BrokerMessage bmsg(pool.mp());
   bmsg.setRecipient(state.operational_recipientkeyname);
   bmsg.setSender(state.operational_senderkeyname);
   bmsg.addTypeHeaderNOCRLF("SET_NEW_ACTOR_ALIAS", actorid);
   bmsg.addHeaderLineNOCRLF("New-Name: %s", actoralias);
}

/** Changes the total blocks of an actor.
 * 
 * @param change
 * The change to be applied. Is positive.
 * 
 * @param actorid
 * The ID of the actor whose blocks are to be modified.
 * 
 * @return Zero on success, one on failure. */
int acquireCommonChangeTotalBlocks(int change, const char *actorid)
{
   return acquireModifyBlocks(SERVER_DO_NOTHING, 0, SERVER_REPLACE_BLOCK_COUNT, change, actorid);
}

/** Changes the blocks done of an actor.
 * 
 * @param change
 * The change to be applied. Is positive.
 * 
 * @param actorid
 * The ID of the actor whose blocks are to be modified.
 * 
 * @return Zero on success, one on failure. */
int acquireCommonChangeDoneBlocks(int change, const char *actorid)
{
   return acquireModifyBlocks(SERVER_REPLACE_BLOCK_COUNT, change, SERVER_DO_NOTHING, 0, actorid);
}

/** Adds to the total blocks of an actor.
 * 
 * @param change
 * The change to be applied. Is positive.
 * 
 * @param actorid
 * The ID of the actor whose blocks are to be modified.
 * 
 * @return Zero on success, one on failure. */
int acquireCommonAddToTotalBlocks(int change, const char *actorid)
{
   return acquireModifyBlocks(SERVER_DO_NOTHING, 0, SERVER_ADD_TO_BLOCK_COUNT, change, actorid);
}

/** Adds to the blocks done of an actor.
 * 
 * @param change
 * The change to be applied. Is positive.
 * 
 * @param actorid
 * The ID of the actor whose blocks are to be modified.
 * 
 * @return Zero on success, one on failure. */
int acquireCommonAddToDoneBlocks(int change, const char *actorid)
{
   return acquireModifyBlocks(SERVER_ADD_TO_BLOCK_COUNT, change, SERVER_DO_NOTHING, 0, actorid);
}

/** Subtracts from the total blocks of an actor.
 * 
 * @param change
 * The change to be applied. Is positive.
 * 
 * @param actorid
 * The ID of the actor whose blocks are to be modified.
 * 
 * @return Zero on success, one on failure. */
int acquireCommonSubtractFromTotalBlocks(int change, const char *actorid)
{
   return acquireModifyBlocks(SERVER_DO_NOTHING, 0, SERVER_SUBTRACT_FROM_BLOCK_COUNT, change, actorid);
}

/** Subtracts from the blocks done of an actor.
 * 
 * @param change
 * The change to be applied. Is positive.
 * 
 * @param actorid
 * The ID of the actor whose blocks are to be modified.
 * 
 * @return Zero on success, one on failure. */
int acquireCommonSubtractFromDoneBlocks(int change, const char *actorid)
{
   return acquireModifyBlocks(SERVER_SUBTRACT_FROM_BLOCK_COUNT, change, SERVER_DO_NOTHING, 0, actorid);
}

/** Returns the time in RFC822 time format. */
const char *acquireCommonGetDateTimeStamp()
{
   return StringUtils::replaceWithPoolString(StringUtils::getDateTimeNowAsString(), state.mp);
}
